package com.ada.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * UpdateStateByKey原语用于记录历史记录，有时，我们需要在 DStream 中跨批次维护状态(例如流计算中累加wordcount)。针对这种情况，updateStateByKey()为我们提供了对一个状态变量的访问，用于键值对形式的 DStream。给定一个由(键，事件)对构成的 DStream，并传递一个指定如何根据新的事件 更新每个键对应状态的函数，它可以构建出一个新的 DStream，其内部数据为(键，状态) 对。
  *
  * updateStateByKey() 的结果会是一个新的 DStream，其内部的 RDD 序列是由每个时间区间对应的(键，状态)对组成的。
  *
  * updateStateByKey操作使得我们可以在用新信息进行更新时保持任意的状态。为使用这个功能，你需要做下面两步：
  * 1、 定义状态，状态可以是一个任意的数据类型。
  * 2、 定义状态更新函数，用此函数阐明如何使用之前的状态和来自输入流的新值对状态进行更新。
  *
  * 使用updateStateByKey需要对检查点目录进行配置，会使用检查点来保存状态。
  */
object SparkStreaming06_UpdateState {

    def main(args: Array[String]): Unit = {

        //创建SparkConf并初始化SSC
        val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming06_UpdateState")
        val ssc = new StreamingContext(sparkConf, Seconds(5))

        //设置检查点目录，用于存放存储数据的文件
        ssc.checkpoint("checkpoint")

        //kafka topic
        val topic = "source"

        //通过KafkaUtil创建kafkaDSteam
        val kafkaDSteam: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(
            ssc,
            "hadoop121:2181",
            topic,
            Map(topic -> 3),
            StorageLevel.MEMORY_ONLY
        )

        //将采集的数据进行分解（扁平化）
        val wordStream: DStream[String] = kafkaDSteam.flatMap(t => t._2.split(" "))

        //将数据进行结构的转换，方便统计分析
        val mapStream: DStream[(String, Int)] = wordStream.map((_, 1))

        //将转换后的数据进行聚合处理
        //这个操作不会进行累加（无状态更新）
        //val wordToSumSteam: DStream[(String, Int)] = mapStream.reduceByKey(_+_)
        //这个操作会将所有数据累加统计（有状态更新）
        val stateStream: DStream[(String, Int)] = mapStream.updateStateByKey {
            case (seq, buffer) => {
                val sum = buffer.getOrElse(0) + seq.sum
                Option(sum)
            }
        }

        //打印结果
        stateStream.print()

        //启动采集器
        ssc.start()
        //Drvier等待采集器的执行
        ssc.awaitTermination()
    }

}

